package chapter06

/**
 * author: yuhui
 * date: 2024 - 11 - 03 8:50 上午
 * descriptions: 广播变量
 * 1）不使用广播变量,使用join算子，有Shuffle
 * 2）不使用广播变量,使用map函数和模式匹配，没有Shuffle
 * 3）使用广播变量,使用map函数和模式匹配，没有Shuffle
 *
 * 运行结果
 * (a,(1,4))
 * (b,(2,5))
 * (c,(3,6))
 * (d,(4,7))
 */

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object Broadcast01 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("BroadcastDemo")
    val sc: SparkContext = new SparkContext(conf)
    val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("b", 2), ("c", 3), ("d", 4)), 4)

    println("==========不使用广播变量,有Shuffle=====分割线===================")
    val list2: List[(String, Int)] = List(("a", 4), ("b", 5), ("c", 6), ("d", 7))

    // 将List转换为RDD
    val rdd2: RDD[(String, Int)] = sc.parallelize(list2)

    // 将RDD转换为键值对RDD，键为String，值为Int
    val keyValueRdd1: RDD[(String, Int)] = rdd.mapValues(v => v) // 实际上这个mapValues是多余的，因为rdd1已经是(String, Int)类型
    val keyValueRdd2: RDD[(String, Int)] = rdd2.mapValues(v => v) // 同上

    // 直接进行连接，不需要转换值的类型
    keyValueRdd2.join(keyValueRdd1).mapValues {
      case (value1, value2) => (value2, value1)
    }.sortBy(_._1).collect().foreach(println)


    println("==========不使用广播变量,没有Shuffle=====分割线=================")
    val map = mutable.Map(("a", 4), ("b", 5), ("c", 6), ("d", 7))
    rdd.map {
      case (key, num) =>
        val l: Int = map.getOrElse(key, 0)
        (key, (num, l))
    }.collect.foreach(println)


    println("==========使用广播变量,没有Shuffle=====分割线===================")
    val list: List[(String, Int)] = List(("a", 4), ("b", 5), ("c", 6), ("d", 7))
    val broadcast: Broadcast[List[(String, Int)]] = sc.broadcast(list)
    rdd.map {
      case (key, num) =>
        var num2 = 0
        for ((k, v) <- broadcast.value) {
          if (k == key) {
            num2 = v
          }
        }
        (key, (num, num2))
    }.collect().foreach(println)


  }
}
